package com.shujia.cep

import org.apache.flink.api.scala._
import org.apache.flink.cep.pattern.conditions.IterativeCondition
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object Demo2CEP {
  def main(args: Array[String]): Unit = {
    /**
      * 1,100,a
      * 1,44,a
      * 1,66,a
      * 1,100,b
      *
      * 如果同一个id连续两次出现了fail   报警
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    val ds = env.socketTextStream("node1", 8888)

    val eventDS = ds.map(_.split(","))
      .map(arr => Event(arr(0).toInt, arr(1).toInt, arr(2)))

    /**
      * 定义规则
      */

    val pattern = Pattern

      .begin[Event]("start") //第一个事件
      .where(event => event.msg.equals("a")) // 条件
      .oneOrMore
      .next("next")
      .where(new IterativeCondition[Event] {
        override def filter(t: Event, context: IterativeCondition.Context[Event]): Boolean = {
          val iter = context.getEventsForPattern("start").iterator()
          println(t)
          var sum = t.num
          while (iter.hasNext) {
            val event = iter.next()
            sum += event.num
          }
          println(sum)

          sum > 200
        }
      })

      .next("end")
      .where(event => event.msg.equals("b"))


    //使用规则在流上进行匹配
    val patternStream = CEP.pattern(eventDS.keyBy(_.id), pattern)


    //如果匹配上  就可以取出匹配成功的数据
    patternStream.select(map => {
      val start = map.getOrElse("start", null)
      val next = map.getOrElse("next", null)
      val end = map.getOrElse("end", null)

      println(start)
      println(next)
      println(end)

    })


    env.execute()


  }

  case class Event(id: Int, num: Int, msg: String)

}
